CF-1871 : Add Catalog Update Command by paras-negi-flink · Pull Request #3304 · confluentinc/cli

Paras Negi (paras-negi-flink) · 2026-04-03T09:01:02Z

Release Notes

Breaking Changes

None

New Features

Added confluent flink catalog update command to update CMF KafkaCatalog on Confluent Platform.

Bug Fixes

None

Checklist

What

This PR implements CF-1871 — CLI: Update SQL catalog for the Confluent CLI, targeting Confluent Platform / CP Flink (CMF on-prem):

Adds a new command group under confluent flink catalog:
- confluent flink catalog update <resourceFilePath>
Wires this command to the existing CMF Catalog REST APIs:
PUT /cmf/api/v1/catalogs/kafka/{catName}.
Updates CmfRestClient wrapper for Catalog operations and corresponding local types/output formatting, following existing patterns used for catalogs and compute pools.

Blast Radius

Scope is limited to new Flink catalog update command; existing CLI behavior (including other confluent flink and confluent kafka commands) is unchanged.
If something goes wrong:
- Impact is confined to users updating Kafka Catalog via the new command:
  - Catalog update may fail or surface CMF errors.
  - Catalog update may not show correct state for databases.
- No changes are made to Flyway migrations or catalog persistence; CMF remains the source of truth.
There are no breaking changes to existing commands, flags, or APIs. Removing or reverting this feature is straightforward (command group + client wrapper).

References

Jira: CF-1871 — CLI: Update SQL catalog
Epic: CF-3136 — 2026 Q1: Streamline CLI with CMF 2.3.0 APIs
CMF migration context (KafkaCatalog → KafkaDatabase): cp-flink-cmf PR that introduced KafkaDatabase and the Flyway migration (e.g., [CF-1772] Add flyway migration for KafkaCatalog -> KafkaDatabase).

Test & Review

Environment

Repo: confluentinc/cli
Branch: CF-1871
CMF: 2.3-SNAPSHOT (image: confluentinc/cp-cmf:c505ee8b) - Kubernetes: local cluster with CMF deployed (cmf-service exposed via kubectl port-forward svc/cmf-service 8080:80 -n e2e)

parasnegi@C6V9RN9V2Y confluent_darwin_arm64_v8.0 % ./confluent flink catalog describe new-kcat --url http://localhost:8080 --output json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "KafkaCatalog",
  "metadata": {
    "name": "new-kcat",
    "creationTimestamp": "2026-04-03T14:46:29.297Z",
    "uid": "41acaae2-62f0-4d00-a8db-4235701ac37e",
    "labels": {},
    "annotations": {}
  },
  "spec": {
    "srInstance": {
      "connectionConfig": {
        "url": "http://localhost:8081"
      }
    },
    "kafkaClusters": []
  }
}

parasnegi@C6V9RN9V2Y confluent_darwin_arm64_v8.0 % ./confluent flink catalog update create-catalog.json --url http://localhost:8080 
+---------------+--------------------------+
| Creation Time | 2026-04-03T14:46:29.297Z |
| Name          | new-kcat                 |
| Databases     |                          |
+---------------+--------------------------+

parasnegi@C6V9RN9V2Y confluent_darwin_arm64_v8.0 % ./confluent flink catalog describe new-kcat --url http://localhost:8080 --output json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "KafkaCatalog",
  "metadata": {
    "name": "new-kcat",
    "creationTimestamp": "2026-04-03T14:46:29.297Z",
    "uid": "41acaae2-62f0-4d00-a8db-4235701ac37e",
    "labels": {},
    "annotations": {}
  },
  "spec": {
    "srInstance": {
      "connectionConfig": {
        "url": "http://localhost:8082"
      }
    },
    "kafkaClusters": []
  }
}

Non-empty KafkaCatalog.spec.kafkaClusters is resricted owing to the Flyway Migration

parasnegi@C6V9RN9V2Y confluent_darwin_arm64_v8.0 % ./confluent flink catalog update update-catalog.json --url http://localhost:8080 
Error: failed to update Kafka Catalog "kcat": {"errors":[{"message":"Catalog 'kcat' not updated. Please do not configure kafkaClusters in the Catalog spec to specify Kafka-backed databases. This field is no longer supported. Use the dedicated KafkaDatabase resource instead."}]}

confluent-cla-assistant · 2026-04-03T09:01:14Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

Copilot

Pull request overview

Adds support for updating CMF Kafka Catalogs via the confluent flink catalog update command (Confluent Platform/on-prem), including REST client support and integration test coverage with golden fixtures.

Changes:

Introduces flink catalog update <resourceFilePath> command (JSON/YAML resource file) and shared catalog output formatting.
Adds CMF REST client support for updating Kafka catalogs and extends the on-prem test server to handle PUT requests.
Adds on-prem integration tests plus input/output fixtures and help goldens for the new command.

Reviewed changes

Copilot reviewed 15 out of 15 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
`internal/flink/command_catalog.go`	Registers the new update subcommand; adds helpers for printing catalog output and reading JSON/YAML resource files.
`internal/flink/command_catalog_update.go`	Implements the `flink catalog update` command behavior (read file → update → describe → print).
`pkg/flink/cmf_rest_client.go`	Adds `UpdateCatalog` wrapper around the CMF SDK update endpoint with CLI-friendly error wrapping.
`test/test-server/flink_onprem_handler.go`	Extends the CMF catalog handler to support PUT for update scenarios.
`test/flink_onprem_test.go`	Adds integration test cases for catalog update (JSON + YAML).
`test/fixtures/input/flink/catalog/*`	Adds JSON/YAML resource files for update success and failure cases.
`test/fixtures/output/flink/catalog/*`	Adds golden outputs for update (human/json/yaml), update failure, and update help; updates catalog help to include `update`.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-04-03T09:04:58Z

+func printCatalogOutput(cmd *cobra.Command, sdkOutputCatalog cmfsdk.KafkaCatalog) error {
+	if output.GetFormat(cmd) == output.Human {
+		table := output.NewTable(cmd)
+		databases := make([]string, 0, len(sdkOutputCatalog.Spec.GetKafkaClusters()))
+		for _, kafkaCluster := range sdkOutputCatalog.Spec.GetKafkaClusters() {
+			databases = append(databases, kafkaCluster.DatabaseName)
+		}
+		var creationTime string
+		if sdkOutputCatalog.GetMetadata().CreationTimestamp != nil {
+			creationTime = *sdkOutputCatalog.GetMetadata().CreationTimestamp
+		}
+		table.Add(&catalogOut{
+			CreationTime: creationTime,
+			Name:         sdkOutputCatalog.GetMetadata().Name,
+			Databases:    databases,
+		})
+		return table.Print()
+	}
+
+	localCatalog := convertSdkCatalogToLocalCatalog(sdkOutputCatalog)
+	return output.SerializedOutput(cmd, localCatalog)
+}


printCatalogOutput centralizes catalog rendering, but catalogCreate and catalogDescribe still contain their own (nearly identical) output formatting logic. To avoid future drift between commands, consider switching those commands to call printCatalogOutput as well.

Robert Metzger (@rmetzger) Kumar Mallikarjuna (@kumar-mallikarjuna) does it make sense to touch other existing files to reduce duplication, or should we tackle that in another change ?

I think it makes sense to fix this as part of this PR

Agree with Copilot and Robert Metzger (@rmetzger) 's comment, maybe we can apply this idea to the catalog database as well.

sonarqube-confluent · 2026-04-03T09:25:23Z

Quality Gate passed

Issues
5 New issues
0 Fixed issues
0 Accepted issues

Measures
0 Security Hotspots
86.2% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube

Robert Metzger (rmetzger) · 2026-04-07T14:06:45Z

+  "apiVersion": "",
+  "kind": "",


Why are apiVersion and kind empty?

Robert Metzger (rmetzger) · 2026-04-07T14:06:58Z

+apiVersion: ""
+kind: ""


this seems off

Channing Dong (channingdong) · 2026-04-14T05:50:53Z

+apiVersion: ""
+kind: ""


Please fill these fields, and similar golden files.

Channing Dong (channingdong) · 2026-04-14T05:52:26Z

+      {
+        "databaseName": "test-database-2",
+        "connectionConfig": null
+      }


question: Are the connectionConfig designed to be null or empty map?

Channing Dong (channingdong) · 2026-04-14T05:54:11Z

+func printCatalogOutput(cmd *cobra.Command, sdkOutputCatalog cmfsdk.KafkaCatalog) error {
+	if output.GetFormat(cmd) == output.Human {
+		table := output.NewTable(cmd)
+		databases := make([]string, 0, len(sdkOutputCatalog.Spec.GetKafkaClusters()))
+		for _, kafkaCluster := range sdkOutputCatalog.Spec.GetKafkaClusters() {
+			databases = append(databases, kafkaCluster.DatabaseName)
+		}
+		var creationTime string
+		if sdkOutputCatalog.GetMetadata().CreationTimestamp != nil {
+			creationTime = *sdkOutputCatalog.GetMetadata().CreationTimestamp
+		}
+		table.Add(&catalogOut{
+			CreationTime: creationTime,
+			Name:         sdkOutputCatalog.GetMetadata().Name,
+			Databases:    databases,
+		})
+		return table.Print()
+	}
+
+	localCatalog := convertSdkCatalogToLocalCatalog(sdkOutputCatalog)
+	return output.SerializedOutput(cmd, localCatalog)
+}


Agree with Copilot and Robert Metzger (@rmetzger) 's comment, maybe we can apply this idea to the catalog database as well.

Channing Dong (channingdong) · 2026-04-14T05:55:30Z

+		return fmt.Errorf("catalog %q was updated successfully, but failed to retrieve updated details: %w", catalogName, err)
+	}
+
+	return printCatalogOutput(cmd, sdkOutputCatalog)


Let's use this unified print out function for the create command as well.

Channing Dong (channingdong) · 2026-04-14T05:56:37Z

+		return fmt.Errorf("catalog name is required: ensure the resource file contains a non-empty \"metadata.name\" field")
+	}
+
+	if err := client.UpdateCatalog(c.createContext(), catalogName, sdkCatalog); err != nil {


Same comment as the catalog database, is it the design that Update command doesn't return the updated catalog to make this an atomic operation? I am seeing an additional describe call below.

Channing Dong (channingdong) · 2026-04-14T05:57:25Z

+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json", fixture: "flink/catalog/update-success.golden"},
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json --output json", fixture: "flink/catalog/update-success-json.golden"},
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json --output yaml", fixture: "flink/catalog/update-success-yaml.golden"},
+		// failure scenarios with JSON files
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-invalid-failure.json", fixture: "flink/catalog/update-invalid-failure.golden", exitCode: 1},


Should these be json format files? Or are these totally redundant?

Channing Dong (channingdong) · 2026-04-14T05:57:45Z

 	runIntegrationTestsWithMultipleAuth(s, tests)
 }

+func (s *CLITestSuite) TestFlinkCatalogUpdateWithYAML() {


Suggested change

func (s *CLITestSuite) TestFlinkCatalogUpdateWithYAML() {

func (s *CLITestSuite) TestFlinkCatalogUpdateOnPremWithYAML() {

Channing Dong (channingdong) · 2026-04-14T06:00:13Z

 				return
 			}

 			timeStamp := time.Date(2025, time.March, 12, 23, 42, 0, 0, time.UTC).String()


Again, although not relevant to this PR, please evaluate the time format from server and see if we should use plain string or .Format(time.RFC3339)

Channing Dong (channingdong) · 2026-04-14T06:00:38Z

+func (c *command) newCatalogUpdateCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "update <resourceFilePath>",
+		Short: "Update a Flink catalog.",


Suggested change

Short: "Update a Flink catalog.",

Short: "Update a Flink catalog in Confluent Platform.",

Channing Dong (channingdong) · 2026-04-14T06:01:12Z

+	cmd := &cobra.Command{
+		Use:   "update <resourceFilePath>",
+		Short: "Update a Flink catalog.",
+		Long:  "Update an existing Kafka Catalog in Confluent Platform from a resource file.",


We want to keep it consistent.

Suggested change

Long: "Update an existing Kafka Catalog in Confluent Platform from a resource file.",

Long: "Update an existing Flink catalog in Confluent Platform from a resource file.",

And by the way, is there a convention for the user to know which fields are mutable from an update operation?

Add Catalog Update Command

6571fb1

Copilot AI review requested due to automatic review settings April 3, 2026 09:01

Copilot started reviewing on behalf of Paras Negi (paras-negi-flink) April 3, 2026 09:01 View session

Copilot AI reviewed Apr 3, 2026

View reviewed changes

Paras Negi (paras-negi-flink) changed the title ~~Add Catalog Update Command~~ CF-1871 : Add Catalog Update Command Apr 3, 2026

Paras Negi (paras-negi-flink) marked this pull request as ready for review April 3, 2026 14:41

Paras Negi (paras-negi-flink) requested a review from a team as a code owner April 3, 2026 14:41

Paras Negi (paras-negi-flink) requested review from Kumar Mallikarjuna (kumar-mallikarjuna) and Robert Metzger (rmetzger) April 3, 2026 15:33

Robert Metzger (rmetzger) reviewed Apr 7, 2026

View reviewed changes

Channing Dong (channingdong) requested changes Apr 14, 2026

View reviewed changes

	func (s *CLITestSuite) TestFlinkCatalogUpdateWithYAML() {
	func (s *CLITestSuite) TestFlinkCatalogUpdateOnPremWithYAML() {

	Short: "Update a Flink catalog.",
	Short: "Update a Flink catalog in Confluent Platform.",

	Long: "Update an existing Kafka Catalog in Confluent Platform from a resource file.",
	Long: "Update an existing Flink catalog in Confluent Platform from a resource file.",

		apiVersion: ""
		kind: ""

		apiVersion: ""
		kind: ""

Conversation

Paras Negi (paras-negi-flink) commented Apr 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Release Notes

Checklist

What

Blast Radius

References

Test & Review

Uh oh!

confluent-cla-assistant bot commented Apr 3, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sonarqube-confluent bot commented Apr 3, 2026

Quality Gate passed

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Channing Dong (channingdong) Apr 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Paras Negi (paras-negi-flink) commented Apr 3, 2026 •

edited

Loading

Channing Dong (channingdong) Apr 14, 2026 •

edited

Loading